package spc.webos.queue.ibmmq;

import com.ibm.mq.MQException;
import com.ibm.mq.MQGetMessageOptions;
import com.ibm.mq.MQMessage;
import com.ibm.mq.MQPutMessageOptions;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import java.io.IOException;
import java.util.GregorianCalendar;
import java.util.Hashtable;
import java.util.List;
import org.apache.log4j.Logger;
import spc.webos.constant.AppRetCode;
import spc.webos.exception.AppException;

public class Accessor {
	public static final int MQ_OK = 0;
	public static final int MQ_CONNECTION_BROKER = -1;
	public static final int MQ_TIME_OUT = 1;
	public static final int MQ_UNKNOW_EXCEPTION = 2;
	public static final int CNN_EXCEPTION_SLEEP = 5;
	public static final String QM_NAME = "qmName";
	public static int CCSID = 1208;
	public static boolean info2033 = false;
	static final Logger log = Logger.getLogger(Accessor.class);

	public static String DEF_FILE_NAME = "data";

	static {
		MQException.log = null;
	}

	public static void receive(List<MQCnnPool> cnnpools, String qname,
			MQGetMessageOptions gmo, MQMessage qmsg) throws Exception {
		for (int i = 0; i < cnnpools.size(); ++i) {
			MQCnnPool cnnpool = (MQCnnPool) cnnpools.get(i);
			try {
				receive(cnnpool, qname, gmo, qmsg);
				return;
			} catch (Exception e) {
				if ((e instanceof MQException)
						&& (((MQException) e).reasonCode == 2033))
					throw e;
				log.warn("cnnpool:" + cnnpool.getProps(), e);
				if (i == cnnpools.size() - 1)
					throw e;
			}
		}
		throw new Exception("fail to read msg from Multi channel, qname: "
				+ qname);
	}

	public static void remove(List<MQCnnPool> cnnpools, String qname,
			MQGetMessageOptions gmo, MQMessage qmsg) {
		gmo.options = 0;
		for (int i = 0; i < cnnpools.size(); ++i)
			try {
				receive((MQCnnPool) cnnpools.get(i), qname, gmo, qmsg);
			} catch (Exception localException) {
			}
	}

	public static void receive(MQManager mqm, String qname,
			MQGetMessageOptions gmo, MQMessage qmsg) throws MQException {
		long start = System.currentTimeMillis();
		int failTimes = 0;
		try {
			if ((((gmo.waitInterval < 0) || (gmo.waitInterval >= 10000)))
					&& (log.isDebugEnabled()))
				log.debug("read MQ(" + qname + ")" + mqm.getCurrentProps()
						+ "," + mqm.keepQueue + ",waitInterval:"
						+ gmo.waitInterval);
			else if (log.isTraceEnabled())
				log.trace("read MQ(" + qname + ")" + mqm.getCurrentProps()
						+ "," + mqm.keepQueue + ",waitInterval:"
						+ gmo.waitInterval);
			mqm.connect(1);

			MQQueue queue = mqm.accessQueue(qname, 8193);

			start = System.currentTimeMillis();
			queue.get(qmsg, gmo);
			long end = System.currentTimeMillis();
			if (log.isInfoEnabled())
				log.info("read MQ(8.0.2)("
						+ Thread.currentThread().getName()
						+ ")("
						+ qname
						+ ")"
						+ mqm.getCurrentProps()
						+ "("
						+ mqm.keepQueue
						+ "), cost:"
						+ (end - start)
						+ ", in mq:"
						+ ((qmsg.putDateTime == null) ? -1L : end
								- qmsg.putDateTime.getTimeInMillis())
						+ ", size: "
						+ qmsg.getMessageLength()
						+ ", corId:"
						+ ((qmsg.correlationId == null) ? "" : new String(
								qmsg.correlationId)));
			return;
		} catch (MQException mqex) {
			++failTimes;
			if (mqex.reasonCode != 2033) {
				log.warn("Err to receive qn: " + qname + ", reason:"
						+ mqex.reasonCode + ", fail:" + failTimes);
				mqm.reconnect(1);
				if (failTimes < 2) {
					mqm.closeQueue();
				}
			}
			throw mqex;
		} catch (Exception ex) {
		} finally {
			mqm.closeQueue();
		}
	}

	public static void receive(MQCnnPool cnnpool, String qname,
			MQGetMessageOptions gmo, MQMessage qmsg) throws Exception {
		if (log.isDebugEnabled())
			log.debug("MQ cnn: " + cnnpool.getProps());
		MQManager mqm = (MQManager) cnnpool.borrow();
		try {
			receive(mqm, qname, gmo, qmsg);
		} finally {
			cnnpool.release(mqm);
		}
	}

	public static void sendAll(List<MQCnnPool> cnnpools, String qname,
			MQPutMessageOptions pmo, MQMessage mqmsg, int retryTimes,
			int retryInterval) throws Exception {
		int index = (int) (Math.random() * 1000000.0D) % cnnpools.size();
		boolean fail = true;
		for (int i = 0; i < cnnpools.size(); ++i) {
			if (index >= cnnpools.size())
				index = 0;
			MQCnnPool cnnpool = (MQCnnPool) cnnpools.get(index++);
			try {
				send(cnnpool, qname, pmo, mqmsg, retryTimes, retryInterval);
				fail = false;
			} catch (Exception e) {
				log.warn("fail to put:" + cnnpool.getProps() + ", qname:"
						+ qname, e);
			}
		}
		if (fail)
			throw new Exception("Fail to sendAll ClusterMQPut: qname:" + qname);
	}

	public static void send(MQCnnPool cnnpool, String qname,
			MQPutMessageOptions pmo, MQMessage mqmsg, int retryTimes,
			int retryInterval) throws Exception {
		MQManager mqm = (MQManager) cnnpool.borrow();
		try {
			send(mqm, qname, pmo, mqmsg, retryTimes, retryInterval);
		} finally {
			cnnpool.release(mqm);
		}
	}

	public static void send(MQManager mqm, String qname,
			MQPutMessageOptions pmo, MQMessage mqMsg, int retryTimes,
			int retryInterval) throws MQException {
		pmo.options |= 1024;
		int opt = 1040;
		if ((pmo.options & 0x800) > 0) {
			opt |= 2048;
		}

		Integer ccsid = (Integer) mqm.getCurrentProps().get("CCSID");
		if (ccsid == null)
			ccsid = Integer.valueOf(CCSID);
		long start = System.currentTimeMillis();
		long failTimes = 0L;
		while (true)
			try {
				mqm.connect(1);

				MQQueue queue = mqm.accessQueue(qname, opt);

				if (mqMsg.characterSet == 0)
					mqMsg.characterSet = ccsid.intValue();
				if (log.isDebugEnabled())
					log.debug("start to put MQ(" + qname + ")"
							+ mqm.getCurrentProps() + ",len:"
							+ mqMsg.getTotalMessageLength());
				queue.put(mqMsg, pmo);
				if (log.isInfoEnabled())
					log.info("put MQ(8.0.5)(" + qname + ")"
							+ mqm.getCurrentProps() + "(" + mqm.keepQueue
							+ "), cost:" + (System.currentTimeMillis() - start)
							+ ", len:" + mqMsg.getTotalMessageLength()
							+ ", retry(" + retryTimes + "," + retryInterval
							+ "," + failTimes + ")" + mqm.name);
				return;
			} catch (MQException mqex) {
				failTimes += 1L;
				int ret = handleMQException(mqex);
				log.warn("err to snd MQ: " + qname + ", reason: "
						+ mqex.reasonCode + ", qm: " + mqm.getCurrentProps()
						+ ", retryTms:" + retryTimes + ", failTms:" + failTimes
						+ ", ret: " + ret + ", cost:"
						+ (System.currentTimeMillis() - start));

				if (ret == -1) {
					log.error("MQ reconnect...");
					mqm.disconnect();
				}
				if (failTimes > retryTimes)
					throw mqex;
				try {
					if (retryInterval > 0)
						Thread.sleep(retryInterval);
				} catch (Exception localException) {
				}
			} finally {
				mqm.closeQueue();
			}
	}

	public static boolean clearAll(MQCnnPool cnnpool, String qname, int maxDepth)
			throws Exception {
		MQManager mqm = (MQManager) cnnpool.borrow();
		MQQueue queue = null;
		try {
			mqm.connect(1);
			MQMessage qmsg = new MQMessage();
			MQGetMessageOptions gmo = new MQGetMessageOptions();
			gmo.matchOptions = 0;
			gmo.options = 0;
			queue = mqm.accessQueue(qname, 1);
			for (int i = 1; i < maxDepth; ++i) {
				try {
					qmsg.clearMessage();
				} catch (IOException e) {
					qmsg = new MQMessage();
				}
				queue.get(qmsg, gmo);
			}
		} catch (MQException mqe) {
			if (handleMQException(mqe) == -1)
				mqm.disconnect();
			if (mqe.reasonCode == 2033)
				return true;
		} finally {
			mqm.closeQueue();
			cnnpool.release(mqm);
		}
		return false;
	}

	public static int browseAll(MQCnnPool cnnpool, String qname, int maxDepth)
			throws Exception {
		int msgnum = 0;
		MQManager mqm = (MQManager) cnnpool.borrow();
		MQQueue queue = null;
		MQMessage qmsg = new MQMessage();
		MQGetMessageOptions gmo = new MQGetMessageOptions();
		gmo.matchOptions = 0;
		gmo.options = 16;
		try {
			mqm.connect(1);
			queue = mqm.accessQueue(qname, 42);

			int curDepth = queue.getCurrentDepth();
			if (log.isInfoEnabled())
				log.info("Queue: " + qname + ", curDepth: " + curDepth);
			if (curDepth <= 0)
				return 0;
			int depth = ((maxDepth <= 0) || (curDepth <= maxDepth)) ? curDepth
					: maxDepth;
			for (int i = 0; i < depth; ++i) {
				try {
					qmsg.clearMessage();
				} catch (IOException e) {
					qmsg = new MQMessage();
				}
				queue.get(qmsg, gmo);
				gmo.options = 32;

				++msgnum;
			}
		} catch (MQException mqe) {
			if (handleMQException(mqe) == -1)
				mqm.disconnect();
			if (mqe.reasonCode != 2033) {
				log.warn("browseAll:" + qname + ":" + mqe.reasonCode);
				throw new AppException(AppRetCode.PROTOCOL_MQ(), new Object[] {
						qname, String.valueOf(mqe.reasonCode),
						String.valueOf(mqe.completionCode) });
			}
			if (log.isInfoEnabled())
				log.info("After browse Queue: " + qname);
		} finally {
			mqm.closeQueue();
			cnnpool.release(mqm);
		}
		return msgnum;
	}

	public static void browse(MQCnnPool cnnpool, String qname,
			MQGetMessageOptions gmo, MQMessage qmsg) throws Exception {
		MQManager mqm = (MQManager) cnnpool.borrow();
		try {
			browse(mqm, qname, gmo, qmsg);
		} finally {
			cnnpool.release(mqm);
		}
	}

	public static void browse(MQManager mqm, String qname,
			MQGetMessageOptions gmo, MQMessage qmsg) throws MQException {
		MQQueue queue = null;
		try {
			mqm.connect(1);
			gmo.options += 16;
			if (log.isInfoEnabled())
				log.info("start to read MQ:" + qname);
			queue = mqm.accessQueue(qname, 8);
			queue.get(qmsg, gmo);
			if (!log.isInfoEnabled())
				log.info("end to read MQ");
		} catch (MQException mqex) {
			if (handleMQException(mqex) == -1)
				mqm.disconnect();
			throw mqex;
		} finally {
			mqm.closeQueue();
		}
	}

	public static MQQueueManager connect(MQQueueManager qm, Hashtable props,
			int retryTimes) {
		long i = 0L;
		String qnname = (String) props.get("qmName");
		while (true) {
			try {
				disconnect(qm);
				qm = new MQQueueManager(qnname, props);
				return qm;
			} catch (Exception e) {
				log.warn(
						"MQ access error:" + props + ", failtimes:" + i
								+ ", sleep:" + 5 + " seconds, retryTimes:"
								+ retryTimes, e);
				i += 1L;
				if ((retryTimes >= 0) && (i >= retryTimes)) {
					if (e instanceof MQException) {
						MQException mqex = (MQException) e;
						throw new AppException(AppRetCode.PROTOCOL_MQ(),
								new Object[] { qnname,
										String.valueOf(mqex.reasonCode),
										String.valueOf(mqex.completionCode),
										e.toString() });
					}
					throw new RuntimeException(e);
				}
				try {
					log.debug("thread will sleep:5");
					Thread.sleep(5000L);
				} catch (Exception localException1) {
				}
			}
			if (retryTimes >= 0)
				continue;
		}
	}

	public static MQQueueManager disconnect(MQQueueManager qm) {
		if (qm == null)
			return null;
		try {
			qm.close();
		} catch (Throwable localThrowable) {
		}
		try {
			qm.disconnect();
		} catch (Throwable localThrowable1) {
		}
		return null;
	}

	public static int handleMQException(MQException e) {
		if (e.completionCode == 0)
			return 0;
		return (e.reasonCode == 2033) ? 1 : -1;
	}
}
